import 'dart:async';

import 'package:rxdart/rxdart.dart';

import '../../common/test_page.dart';

class ConcatTestPage extends TestPage {
  ConcatTestPage(super.title) {
    test('Rx.concat', () async {
      const expectedOutput = [0, 1, 2, 3, 4, 5];
      var count = 0;

      final stream = Rx.concat(_getStreams());

      stream.listen(((result) {
        // test to see if the combined output matches
        expect('$result, ${expectedOutput[count++]}');
      }));
    });

    test('Rx.concatEager.single.subscription', () async {
      final stream = Rx.concat(_getStreams());

      stream.listen(null);
      expect(() => stream.listen(null));
    });

    test('Rx.concat.withEmptyStream', () async {
      const expectedOutput = [0, 1, 2, 3, 4, 5];
      var count = 0;

      final stream = Rx.concat(_getStreamsIncludingEmpty());

      stream.listen(((result) {
        // test to see if the combined output matches
        expect('$result, ${expectedOutput[count++]}');
      }));
    });

    test('Rx.concat.withBroadcastStreams', () async {
      const expectedOutput = [1, 2, 3, 4];
      final ctrlA = StreamController<int>.broadcast(),
          ctrlB = StreamController<int>.broadcast(),
          ctrlC = StreamController<int>.broadcast();
      var x = 0, y = 100, z = 1000, count = 0;

      Timer.periodic(const Duration(milliseconds: 1), (_) {
        ctrlA.add(++x);
        ctrlB.add(--y);

        if (x <= 3) ctrlC.add(--z);

        if (x == 3) ctrlC.close();

        if (x == 4) {
          _.cancel();

          ctrlA.close();
          ctrlB.close();
        }
      });

      final stream = Rx.concat([ctrlA.stream, ctrlB.stream, ctrlC.stream]);

      stream.listen(((result) {
        // test to see if the combined output matches
        expect('$result, ${expectedOutput[count++]}');
      }));
    });

    test('Rx.concat.asBroadcastStream', () async {
      final stream = Rx.concat(_getStreams()).asBroadcastStream();

      // listen twice on same stream
      stream.listen(null);
      stream.listen(null);
      // code should reach here
      expect(stream.isBroadcast);
    });

    test('Rx.concat.error.shouldThrowA', () async {
      final streamWithError =
      Rx.concat(_getStreams()..add(Stream<int>.error(Exception())));

      streamWithError.listen(null,
          onError: (Object e, StackTrace s) {
            expect(e);
          });
    });

    test('Rx.concat.empty', () {
      expect(Rx.concat<int>(const []));
    });

    test('Rx.concat.single', () {
      expect(
        Rx.concat<int>([Stream.value(1)])
      );
    });

    test('Rx.concat.iterate.once', () async {
      var iterationCount = 0;

      final stream = Rx.concat<int>(() sync* {
        ++iterationCount;
        yield Stream.value(1);
        yield Stream.value(2);
        yield Stream.value(3);
      }());

      expect(
        stream
      );
      expect(iterationCount);
    });
  }
  List<Stream<int>> _getStreams() {
    var a = Stream.fromIterable(const [0, 1, 2]),
        b = Stream.fromIterable(const [3, 4, 5]);

    return [a, b];
  }

  List<Stream<int>> _getStreamsIncludingEmpty() {
    var a = Stream.fromIterable(const [0, 1, 2]),
        b = Stream.fromIterable(const [3, 4, 5]),
        c = Stream<int>.empty();

    return [c, a, b];
  }
}